package net.bwie.douyin.dwd.log.Function;


import net.bwie.douyin.dwd.log.bean.LiveEvent;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class GiftAbnormalFunction implements WindowFunction<LiveEvent, String, String, TimeWindow> {

    private FastDateFormat format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");


    @Override
    public void apply(String s, TimeWindow window, Iterable<LiveEvent> input, Collector<String> out) throws Exception {
        String windowStartTime = format.format(window.getStart());
        String windowEndTime = format.format(window.getEnd());


        int gift_sum = 0;
        for (LiveEvent liveEvent : input) {
            gift_sum+= liveEvent.getGiftValue();
        }

        if(gift_sum>=1000){
            out.collect(windowStartTime + "," + windowEndTime + "," + s  + "," + gift_sum+","+System.currentTimeMillis());
     }

    }
}
